package com.lry.util.mysql;

import com.alibaba.excel.EasyExcel;
import com.lry.util.mysql.util.SqlUtil;
import java.io.BufferedReader;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;

/**
 * @author ALI
 * @since 2023/3/7
 */
@Slf4j
public class FileToDbUtil {
    public static final int LATCH_COUNT = 500;
    public static final int MAX_LINE = 10000;
    private static final String CREATE_STR = "create";
    private static final String CREATE_STR2 = "CREATE";
    private static final String LINE_SEPARATOR = ";";
    private static final String LC = "\r\n";
    private static ExecutorService pool = null;

    static {
        // 手动创建线池，给定有界线程队列
        pool = new ThreadPoolExecutor(4, 6,
                                      0L, TimeUnit.MILLISECONDS,
                                      new LinkedBlockingQueue<Runnable>(200));
    }

    /**
     * 读取sql文件，每读取10000条数据，执行option方法
     * 注意：
     * 1. 对于大数据量，必须通过option方法进行数据释放
     * 2. 分批次的数据导入，建议每次都重新获取connection
     * 文件格式:
     * create table xx();\r\n
     * insert into xx();
     * ......
     *
     * @param inputStream 文件流
     * @param consumer    数据库执行方法对象
     * @return 建表sql
     */
    public static String readSqlFile(InputStream inputStream, Consumer<String> consumer) throws IOException {
        boolean start = false;
        boolean end = false;
        boolean create = false;
        int lines = 0;
        String createSql = null;
        InputStreamReader ir = new InputStreamReader(inputStream);
        StringBuilder insertSb = new StringBuilder();
        StringBuilder createSb = new StringBuilder();
        CountDownLatch latch = new CountDownLatch(LATCH_COUNT);
        List<Thread> threads = new ArrayList<>();
        try (BufferedReader is = new BufferedReader(ir)) {
            String s = null;
            while ((s = is.readLine()) != null) {
                if (++lines > MAX_LINE) {
                    String sql = insertSb.toString();
                    if (!create) {
                        create = true;
                    }
                    threads.add(new Thread(() -> {
                        executeSql(sql, consumer);
                        latch.countDown();
                    }));
                    insertSb = new StringBuilder();
                    lines = 0;
                }
                if (s.contains(CREATE_STR) || s.contains(CREATE_STR2)) {
                    start = true;
                    if (s.contains(LINE_SEPARATOR)) {
                        createSb.append(s, 0, s.indexOf(LINE_SEPARATOR));
                    } else {
                        createSb.append(s);
                    }
                    createSb.append(LC);
                } else if (start) {
                    if (!end) {
                        if (s.contains(LINE_SEPARATOR)) {
                            createSb.append(s, 0, s.indexOf(LINE_SEPARATOR));
                            createSql = executorCreateTableSql(consumer, createSb.toString());
                            is.mark(1);
                            end = true;
                        } else {
                            createSb.append(s);
                        }
                        createSb.append(LC);
                    } else {
                        insertSb.append(s).append(LC);
                    }
                }
            }
            String insertSql = insertSb.toString();
            threads.add(new Thread(() -> {
                executeSql(insertSql, consumer);
                latch.countDown();
            }));
            log.info("数据入库（db import）- 等待子线程执行......");
            waitThread(threads, latch);
            log.info("数据入库（db import）- 子线程执行完毕......");
        }
        return createSql;
    }

    /**
     * 执行sql
     *
     * @param sql    sql
     * @param option 执行操作
     */
    private static void executeSql(String sql, Consumer<String> option) {
        if (StringUtils.isNotBlank(sql) && option != null) {
            option.accept(sql);
        }
    }

    /**
     * 执行并获取建表sql；自动添加drop table is exist;
     */
    private static String executorCreateTableSql(Consumer<String> option, String createSql) {
        createSql = SqlUtil.toExcutorSql(createSql);
        List<TableInfo> tableInfos = SqlUtil.parseCreateSql(createSql, "1");
        TableInfo tableInfo = tableInfos.get(0);
        String dropSql = String.format("DROP TABLE IF EXISTS `%s`;", tableInfo.getNewName());
        executeSql(dropSql + createSql, option);
        return createSql;
    }

    private static void waitThread(List<Thread> threads, CountDownLatch latch) {
        for (Thread thread : threads) {
            pool.execute(thread);
        }
        while (true) {
            try {
                TimeUnit.MILLISECONDS.sleep(20);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            if ((LATCH_COUNT - latch.getCount()) == threads.size()) {
                return;
            }
        }
    }

    /**
     * 从excel文件中解析出sql和表信息
     */
    private void parseExcelFile(InputStream inputStream, Consumer<Map<Integer, String>> headerConsumer,
                                Consumer<ByteArrayOutputStream> bodyConsumer) throws IOException {
        log.info("数据入库（db import）- 开始解析excel文件......");
        ReadExcelListener listener = new ReadExcelListener(headerConsumer, bodyConsumer);
        EasyExcel.read(inputStream, listener).headRowNumber(0).doReadAll();
        listener.waitThread();
        log.info("数据入库（db import）- 读取excel总数据行数：{}", listener.getTotalLines());
        log.info("数据入库（db import）- excel文件导入成功！");
    }
}
